package com.pipi.mq.rocketmq.base.producer;

import com.alibaba.fastjson.JSON;
import com.pipi.mq.rocketmq.base.config.PropertiesConfig;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.util.Date;
import java.util.concurrent.*;

/**
 * @Package com.pipi.mq.rocketmq.base
 * @author: 皮皮（pipi@maihaoche.com）
 * @date: 2020/4/18 1:54 PM
 */
@Component
public class AsyncProducer extends AbstractMQProducer {

    private static int n = 1001;

    public AsyncProducer() {
        setTopic(PropertiesConfig.topicAsync);
        setTag(PropertiesConfig.tagAsync);
    }

    public void asyncSendThread(CountDownLatch countDownLatch) {
        ExecutorService executorService = new ThreadPoolExecutor(1,
                1,
                100L,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(10));
        executorService.submit(() -> {
            asyncSend();
            countDownLatch.countDown();
        });
        executorService.shutdown();
    }

    public void asyncSend() {
        CountDownLatch countDownLatch = new CountDownLatch(n);
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < n; i++) {
            String s = DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss") + "  asyncSendMessage-" + i;
            asynSend("", s, s, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println(JSON.toJSONString(sendResult));
                    countDownLatch.countDown();
                }
                @Override
                public void onException(Throwable throwable) {
                    System.out.println(throwable.getMessage());
                    throwable.printStackTrace();
                    countDownLatch.countDown();
                }
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(System.currentTimeMillis() - startTime);
    }


}
